package callback

import (
	"errors"
	"sync"
	"time"
)

// Wait 等待机制
type Wait struct {
	m       map[string]*wait       //map
	mData   map[string]interface{} //数据,共享
	mu      sync.RWMutex           //锁
	timeOut time.Duration          //超时时间
	reuse   bool                   //复用模式
}

func NewWait(timeOut time.Duration) *Wait {
	return &Wait{
		m:       make(map[string]*wait),
		mData:   make(map[string]interface{}),
		timeOut: timeOut,
		reuse:   false,
	}
}

// SetReuse 设置复用,同个key,响应的参数是否复用
// 如果设置复用,同时2个相同的key等待数据,那么2个等待都会释放
func (this *Wait) SetReuse(b ...bool) *Wait {
	this.reuse = !(len(b) > 0 && !b[0])
	return this
}

// SetTimeout 设置超时时间
func (this *Wait) SetTimeout(t time.Duration) *Wait {
	this.timeOut = t
	return this
}

func (this *Wait) Wait(key string) (*Msg, error) {
	this.mu.RLock()
	w, ok := this.m[key]
	this.mu.RUnlock()

	if ok {
		if this.reuse {
			data := <-w.result()
			return data.Data, data.Error
		} else {
			timer := time.NewTimer(this.timeOut)
			select {
			case <-w.finish():
				timer.Stop()
			case <-timer.C:
				return nil, errors.New("超时")
			}
		}
	}

	w = newWait(this.timeOut)
	this.mu.Lock()
	this.m[key] = w
	this.mu.Unlock()

	data, err := w.wait()

	this.mu.Lock()
	defer this.mu.Unlock()
	delete(this.m, key)
	return data, err
}

// Done 结束等待
// @key,等待唯一标识符
// @v,等待的数据
func (this *Wait) Done(key string, v *Msg) bool {
	this.mu.Lock()
	defer this.mu.Unlock()
	_, ok := this.m[key]
	if ok {
		this.m[key].done(v)
		delete(this.m, key)
	}
	return ok
}

//===================================================================//

//数据包
type data struct {
	Data  *Msg  //数据
	Error error //错误
}

func newData(v *Msg, err error) *data {
	return &data{
		Data:  v,
		Error: err,
	}
}

//等待机制,消费一次
type wait struct {
	c       chan *Msg     //等待通道
	timeOut time.Duration //超时时间
	_finish chan struct{} //结束
	_result []chan *data  //结果
}

func newWait(timeOut time.Duration) *wait {
	return &wait{
		c:       make(chan *Msg, 1),
		timeOut: timeOut,
		_finish: make(chan struct{}, 1),
	}
}

func (this *wait) finish() <-chan struct{} {
	return this._finish
}

func (this *wait) result() <-chan *data {
	c := make(chan *data)
	this._result = append(this._result, c)
	return c
}

func (this *wait) reResult() {
	this._result = []chan *data{}
}

//等待回调
func (this *wait) wait() (data *Msg, err error) {
	timer := time.NewTimer(this.timeOut)
	defer func() {
		timer.Stop()
		if len(this._finish) == 0 {
			this._finish <- struct{}{}
		}
		for _, v := range this._result {
			v <- newData(data, err)
			close(v)
		}
		this.reResult()
	}()
	select {
	case v := <-this.c:
		data = v
	case <-timer.C:
		err = errors.New("超时")
	}
	return
}

//收到回调,结束等待
func (this *wait) done(v *Msg) {
	if len(this.c) == 0 {
		this.c <- v
	}
}
